package com.taobao.analyse.reduce;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.springframework.util.CollectionUtils;

import com.taobao.analyse.constants.SplitConstants;
import com.taobao.cmp.citadel.message.PropConstants;
import com.taobao.item.constant.ItemSkuUpdateChangedField;
import com.taobao.item.domain.ItemSkuDO;

public class ItemSourceReducer extends MapReduceBase implements
		Reducer<LongWritable, Text, LongWritable, Text> {

	@Override
	public void reduce(LongWritable key, Iterator<Text> values,
			OutputCollector<LongWritable, Text> output, Reporter reporter)
			throws IOException {
		String price = "";
		List<String> skus = new ArrayList<String>();
		while(values.hasNext()){
			String data = values.next().toString();
			if(data.startsWith("price")){
				price = data;
			}else{
				skus.add(data);
			}
		}
		StringBuilder sb = new StringBuilder();
		this.buildSourceRecord(sb, price, skus);
		String result = sb.toString();
		if(StringUtils.isEmpty(result)){
			return;
		}
		output.collect(key, new Text(result));
	}

	/**
	 * 生成编辑价格的记录，用于二次分析
	 * 封装格式：source@price:1234;sku:0_0,1_100,2_200,
	 */
	public void buildSourceRecord(StringBuilder sb,
			String price, List<String> skus) {
		
		sb.append("source").append(SplitConstants.SPLIT_AT);
		sb.append(price);
		sb.append(SplitConstants.SPLIT_SEMCOLON);
		buiidSkus(sb,skus);
	}

	/**
	 * 生成sku变更的组装记录
	 * 如果没有一个sku的价格进行变更，则输出：sku:null
	 * 若有，则输出：sku:skuId1_skuPrice1,skuId2_skuPrice2,
	 * @param sb
	 * @param skus
	 */
	private void buiidSkus(StringBuilder sb, List<String> skus) {
		//生成sku:
		sb.append(PropConstants.SKU).append(SplitConstants.SPLIT_COLON);
		if(CollectionUtils.isEmpty(skus)){
			sb.append("null");
		}else{
			for(String sku : skus){
					sb.append(sku);
					sb.append(SplitConstants.SPLIT_COMMA);
			}
		}
	}
	

}
